package com.ds.lens.flink.monitor.common;

import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.util.Preconditions;

import java.util.Arrays;

/**
 * describe:
 *
 * @author wangyiming
 * @date 2019/07/06
 */
public class KafkaKeyedPartition<T> extends FlinkKafkaPartitioner<T> {

    @Override
    public int partition(T record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
        Preconditions.checkArgument(
                partitions != null && partitions.length > 0,
                "Partitions of the target topic is empty.");
        return partitions[Math.abs(Arrays.hashCode(key)) % partitions.length];
    }

    @Override
    public boolean equals(Object o) {
        return this == o || o instanceof KafkaKeyedPartition;
    }

    @Override
    public int hashCode() {
        return KafkaKeyedPartition.class.hashCode();
    }
}
